
#pragma once
#include "logmessage.hpp"
#include "Socket.hpp"
#include "Any.hpp"
#include <iostream>
#include <vector>
#include <cassert>
#include <cstring>
#include <string>
#include <functional>
#include <sys/epoll.h>
#include <unordered_map>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <sys/eventfd.h>
#include <memory>
#include <sys/timerfd.h>
#include <signal.h>

#define BUFFER_DEFAULT_SIZE 1024

class Buffer
{
private:
    std::vector<char> _buffer; // 使用vector进行内存空间管理
    uint64_t _read_idx;        // 读偏移
    uint64_t _writer_idx;      // 写偏移

public:
    Buffer() : _read_idx(0), _writer_idx(0), _buffer(BUFFER_DEFAULT_SIZE) {}
    char *Begin()
    {
        return &*_buffer.begin();
    }

    // 1.获取当前写入起始地址
    char *WritePosition()
    {
        // buffer的空间起始地址加上写偏移量
        return Begin() + _writer_idx;
    }

    // 2.获取当前读取起始地址
    char *ReadPosition()
    {
        return Begin() + _read_idx;
    }
    // 3.获取空闲空间大小  前沿空间大小   后延空间大小
    uint64_t TailIdleSize()
    {
        // 总体空间大小减去写偏移
        return _buffer.size() - _writer_idx;
    }

    uint64_t HeadIdleSize()
    {
        return _read_idx;
    }

    // 4.获取可读数据大小
    uint64_t ReadAbleSize()
    {
        return _writer_idx - _read_idx;
    }

    // 5.将读偏移向后移动
    void MoveReadOffset(uint64_t len)
    {
        // 向后移动大小必须小于可读数据大小
        if(0 == len){
            return;
        }
        assert(len <= ReadAbleSize());
        _read_idx += len;
    }

    // 6.将写偏移向后移动
    void MoveWriteOffset(uint64_t len)
    {
        assert(len <= TailIdleSize()); // 写入前需要确保后延空间足够
        _writer_idx += len;
    }

    // 7.确保可写空间足够  数据挪动 还是 扩容
    void EnsureWriteSpace(uint64_t len)
    {
        // 1.末尾空闲空间足够 就返回
        if (TailIdleSize() >= len)
        {
            return;
        }
        // 末尾空间不够 加起始空间大小，够就挪动  不够就扩容
        if (len <= TailIdleSize() + HeadIdleSize())
        {
            // 挪动数据
            uint64_t rsz = ReadAbleSize();                            // 获取当前数据大小
            std::copy(ReadPosition(), ReadPosition() + rsz, Begin()); // 可读数据拷贝到了起始位置
            _read_idx = 0;                                            // 读偏移归零
            _writer_idx = rsz;                                        // 写偏移置为可读数据大小即写偏移量
        }
        else // 空间不够 需扩容
        {
            // 不移动数据，直接给写偏移之后扩容足够空间即可
            _buffer.resize(_writer_idx + len);
        }
    }

    // 写入数据
    void Write(const void *data, uint64_t len)
    {
        if(0 == len){
            return;
        }
        // 1.保证空间足够  2.拷贝数据
        EnsureWriteSpace(len);
        const char *d = (const char *)data;
        std::copy(d, d + len, WritePosition());
    }
    void WriteAndPush(const void *data, uint64_t len)
    {
        Write(data, len);
        MoveWriteOffset(len);
    }

    void WriteString(const std::string &data)
    {
        Write(data.c_str(), data.size());
    }
    void WriteStringAndPush(const std::string &data)
    {
        WriteString(data);
        MoveWriteOffset(data.size());
    }

    void WriteBuffer(Buffer &data)
    {
        Write(data.ReadPosition(), data.ReadAbleSize());
    }

    void WriteBufferAndPush(Buffer &data)
    {
        WriteBuffer(data);
        MoveWriteOffset(data.ReadAbleSize());
    }

    // 读取数据
    void Read(void *buf, uint64_t len)
    {
        // 要求获取数据大小小于可读数据大小
        assert(len <= ReadAbleSize());
        std::copy(ReadPosition(), ReadPosition() + len, (char *)buf);
    }
    void ReadAndPop(void *buf, uint64_t len) // 读数据并且弹出已读数据
    {
        Read(buf, len);
        MoveReadOffset(len);
    }

    std::string ReadAsString(uint64_t len)
    {
        assert(len <= ReadAbleSize());
        std::string str;
        str.resize(len);
        Read(&str[0], len);
        return str;
    }
    std::string ReadAsStringAndPop(uint64_t len)
    {
        assert(len <= ReadAbleSize());
        std::string str = ReadAsString(len);
        MoveReadOffset(len);
        return str;
    }

    char *FindCRLF()
    {
        void *res = memchr(ReadPosition(), '\n', ReadAbleSize()); // 在空间中找对应字符
        return (char *)res;
    }

    std::string GetLine()
    {
        char *pos = FindCRLF();
        if (pos == nullptr)
        {
            return "";
        }
        return ReadAsString(pos - ReadPosition() + 1); //+1是为了取出换行字符
    }

    std::string GetLineAndPop()
    {
        std::string str = GetLine();
        MoveReadOffset(str.size());
        return str;
    }

    // 清空缓冲区
    void Clear() // 将偏移量归零即可
    {
        _read_idx = 0;
        _writer_idx = 0;
    }
};

//------------------------------------------------------------------------------------------------------------------------------------------------------

// channel模块
// 对一个描述符进行IO事件管理     实现对描述符可读、可写等的操作
// Poller模块对描述符进行IO 事件监控就绪后，回调不同的处理函数功能
class Poller;
class EventLoop;
class Channel
{
private:
    int _fd;
    uint32_t _event;  // 当前需要监控的事件
    uint32_t _events; // 当前连续触发的事件
    EventLoop *_loop;
    using EventCallBack = std::function<void()>;
    EventCallBack _read_callback;  // 读事件触发回调
    EventCallBack _write_callback; // 写事件触发回调
    EventCallBack _error_callback; // 错误事件触发回调
    EventCallBack _close_callback; // 断开事件触发回调
    EventCallBack _event_callback; // 任意事件触发回调
public:
    Channel(EventLoop *loop, int fd) : _loop(loop), _fd(fd), _event(0), _events(0){};
    int Fd()
    {
        return _fd;
    }
    uint32_t Event()
    {
        return _event;
    }
    // 设置实际就绪实际
    void SetEvents(uint32_t _event)
    {
        _events = _event;
    }
    // 1.是否可读
    bool IsRead()
    {
        return _event & EPOLLIN;
    }
    // 2.是否可写
    bool IsWrite()
    {
        return _event & EPOLLOUT;
    }
    // 3.使 可读
    void EnableRead()
    {
        _event |= EPOLLIN;
        UpDate();
    }
    // 4.使 可写
    void EnableWrite()
    {
        _event |= EPOLLOUT;
        UpDate();
    }
    // 5.取消可读
    void DisableRead()
    {
        _event &= ~EPOLLIN;
        UpDate();
    }
    // 6.取消可写
    void DisableWrite()
    {
        _event &= ~EPOLLOUT;
        UpDate();
    }
    // 7.取消所有监控
    void DisableAll()
    {
        _event = 0;
        UpDate();
    }

    // 8.设置各个回调函数
    void SetWriteCallBack(const EventCallBack &cb)
    {
        _write_callback = cb;
    }
    void SetReadCallBack(const EventCallBack &cb)
    {
        _read_callback = cb;
    }
    void SetErrorCallBack(const EventCallBack &cb)
    {
        _error_callback = cb;
    }
    void SetCloseCallBack(const EventCallBack &cb)
    {
        _close_callback = cb;
    }
    void SetEventCallBack(const EventCallBack &cb)
    {
        _event_callback = cb;
    }

    // 9.事件处理函数
    void HandleEvent()
    {
        if (_event_callback)
        {
            _event_callback();
        }
        if ((_events & EPOLLIN) || (_events & EPOLLRDHUP) || (_events & EPOLLPRI))
        { // 对方关闭连接也触发可读事件
            if (_read_callback){
                _read_callback();
            }
        }
        if (_events & EPOLLOUT)
        {  
            if (_write_callback)
            {
                _write_callback();
            }
        } // 有可能释放连接的操作事件，一次只处理一个
        else if (_events & EPOLLERR)
        {
            if (_error_callback)
            {
            
                _error_callback();
            }
        }
        else if (_events & EPOLLHUP)   
        {   
            if (_close_callback)
            {
                _close_callback();
            }
        }
    }
    void UpDate();
    void ReMove();
    // void UpDate(){
    //     return _poller->UpDateEvent(this);
    // }
    // //移除监控
    // void ReMove(){
    //     return _poller->ReMoveEvent(this);
    // }
};
//------------------------------------------------------------------------------------------------------------------------------------------------------
#define MAX_EPOLLEVENTS 1024
// poller模块
class Poller
{
private:
    int _epfd;
    struct epoll_event _evs[MAX_EPOLLEVENTS];
    std::unordered_map<int, Channel *> _channels;

private:
    // 对epoll的直接操作
    void UpDate(Channel *channel, int op)
    {
        // int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
        int fd = channel->Fd();
        struct epoll_event ep;
        ep.data.fd = fd;
        ep.events = channel->Event();

        int ret = epoll_ctl(_epfd, op, fd, &ep);
        if (ret < 0)
        {
            ERR_LOG("EPOLL CTL ERR");
        }
        return;
    }
    // 判断一个Channel是否添加了监控
    bool HasChannel(Channel *channel)
    {
        auto it = _channels.find(channel->Fd());
        if (it == _channels.end())
        {
            return false;
        }
        return true;
    }

public:
    Poller()
    {
        // int epoll_create(int size);   the size argument is ignored, but must be greater than zero;
        int ret = epoll_create(1);
        if (ret < 0)
        {
            ERR_LOG("EPOLL CREATE ERR");
            abort(); // 创建失败 终止程序
        }
        _epfd = ret;
    }
    // 1.添加或修改监控事件
    void  UpDateEvent(Channel *channel)
    {
        bool ret = HasChannel(channel);
        if (ret == false)
        { // 不存在 添加
            _channels.insert(std::make_pair(channel->Fd(), channel));
            return UpDate(channel, EPOLL_CTL_ADD);
        }
        // 存在更新
        return UpDate(channel, EPOLL_CTL_MOD);
    }
    // 2.移除监控
    void ReMoveEvent(Channel *channel)
    {
        auto it = _channels.find(channel->Fd());
        if (it != _channels.end())
        {
            _channels.erase(it);
        }
        UpDate(channel, EPOLL_CTL_DEL);
    }
    // 3.开始监听 ，返回活跃连接
    void Poll(std::vector<Channel *> *active)
    {
        // int epoll_wait(int epfd, struct epoll_event *events,int maxevents, int timeout);
        int nfds = epoll_wait(_epfd, _evs, MAX_EPOLLEVENTS, -1);
        if (nfds < 0)
        {
            if (errno == EINTR)
            {
                return;
            }
            ERR_LOG("EPOLL WAIT ERROR:%S\n", strerror(errno));
            abort();
        }
        for (int i = 0; i < nfds; ++i)
        {
            auto it = _channels.find(_evs[i].data.fd);
            assert(it != _channels.end());
            it->second->SetEvents(_evs[i].events);
            active->push_back(it->second);
        }
    }
};

using TaskFunc = std::function<void()>;
using ReleaseFunc = std::function<void()>;

//------------------------------------------------------------------------------------------------------------------------------------------------------

// 封装定时任务对象  通过析构函数和shared_ptr实现延迟销毁

class TimerTask
{

private:
    uint64_t _id;         // 标识该任务对象id
    uint32_t _timeout;    // 超时时间
    TaskFunc _task_cb;    // 要执行的定时任务
    ReleaseFunc _release; // 用于删除timewheel中保存的定时任务对象信息
    bool _canceled;       // false 表示未被取消  true，表示取消

public:
    TimerTask(uint64_t id, uint32_t delay, const TaskFunc &cb)
        : _id(id), _timeout(delay), _task_cb(cb), _canceled(false)
    {
    }

    ~TimerTask()
    {
        // 析构的时候再执行定时任务
        if (_canceled == false)
        {
            _task_cb();
            _release();
        }
    }
    void TimerCancle()
    {
        _canceled = true;
    }

    void SetRelease(const ReleaseFunc &re) // 设置
    {
        _release = re;
        return;
    }

    uint32_t DelayTime()
    {
        return _timeout; // 获取延迟时间
    }
};

class EventLoop;
// 时间轮
class TimeWheel
{

private:
    using WeakTask = std::weak_ptr<TimerTask>;
    using PtrTask = std::shared_ptr<TimerTask>;

    int _tick;     // 秒针，指到哪执行到哪里
    int _capacity; // 表盘的最大数量，即最大的延迟时间
                   // 用二维数组存储时间段及该时间的定时任务
    std::vector<std::vector<PtrTask>> _wheel;
    // 建立定时任务id和weakptr的映射关系   涉及share_ptr的缺陷
    std::unordered_map<uint64_t, WeakTask> _timers;

    EventLoop *_loop;
    int _timerfd; // 定时器描述  -- 可读事件回调就是读取计数器 执行定时任务
    std::unique_ptr<Channel> _timer_channel;

private:
    void RemoveTimer(uint64_t id)
    {
        auto it = _timers.find(id);
        if (it != _timers.end()) // 找到该定时任务
        {
            _timers.erase(it);
        }
        return;
    }
    static int CreateTimerfd()
    {
        int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        if (timerfd < 0)
        {
            ERR_LOG("timerfd_create FAILED");
            return -1;
        }
        // int timerfd_settime(int fd, int flags,const struct itimerspec *new_value,struct itimerspec *old_value);
        struct itimerspec itime;
        itime.it_value.tv_sec = 1;
        itime.it_value.tv_nsec = 0;
        itime.it_interval.tv_sec = 1;
        itime.it_interval.tv_nsec = 0;
        timerfd_settime(timerfd, 0, &itime, NULL);
        return timerfd;
    }

    void ReadTimerfd()
    {
        uint64_t times;
        int ret = read(_timerfd, &times, 8);
        if (ret < 0)
        {
            ERR_LOG("READ TIMEFD FAILED");
            abort();
        }
        return;
    }

    void OnTime()
    {
        ReadTimerfd();
        RunTimerTask();
    }
    void TimerAddInLoop(uint64_t id, uint32_t delay, const TaskFunc &cb) // 添加定时任务
    {
        PtrTask pt(new TimerTask(id, delay, cb));
        pt->SetRelease(std::bind(&TimeWheel::RemoveTimer, this, id)); // 为什么要&

        _timers[id] = WeakTask(pt);
        int pos = (_tick + delay) % _capacity; // 循环队列一样
        _wheel[pos].push_back(pt);             // 插入延迟任务
    }
    void TimerCancleInLoop(uint64_t id)
    {
        auto it = _timers.find(id);
        if (it == _timers.end())
        {
            return; // 没该定时任务，返回
        }
        // 找到后调用
        PtrTask pt = it->second.lock();   //获取智能指针的时候，需要判断获取的智能指针是否是有管理对象的，
        if (pt) pt->TimerCancle();
        return;
    }
    void TimerRefreshInLoop(uint64_t id)
    { // 刷新/延迟定时任务
        // 通过保存的定时器对象中的weak_ptr构造一个share_ptr，添加到轮子
        auto it = _timers.find(id);
        if (it == _timers.end())
        {
            return; // 没该定时任务，返回
        }
        PtrTask pt = it->second.lock(); // lock获取weak_ptr管理的对象对应的shared_ptr
        uint32_t delay = pt->DelayTime();
        int pos = (_tick + delay) % _capacity; // 循环队列一样

        _wheel[pos].push_back(pt); // 插入延迟任务
    }

public:
    TimeWheel(EventLoop *loop)
        : _capacity(60), _tick(0), _wheel(_capacity), _loop(loop), _timerfd(CreateTimerfd()), _timer_channel(new Channel(_loop, _timerfd))
    {
        _timer_channel->SetReadCallBack(std::bind(&TimeWheel::OnTime, this));
        _timer_channel->EnableRead(); // 启动读事件监控
    }
    // 定时器的操作有可能在多线程中执行，得保证线程安全
    // 不想加锁，就得把操作都放到一个线程中执行
    void TimerAdd(uint64_t id, uint32_t delay, const TaskFunc &cb);
    void TimerRefresh(uint64_t id);
    void TimerCancle(uint64_t id);
    bool HashTimer(uint64_t id)
    {
        auto it = _timers.find(id);
        if (it == _timers.end())
        {
            return false; // 没该定时任务，返回
        }
        return true;
    }
    void RunTimerTask()
    { // 执行定时任务，应该每秒执行一次，相当于秒针走一步
        _tick = (_tick + 1) % _capacity;
        _wheel[_tick].clear(); // 清空该位置的定时任务，自动调用定时任务对象的析构
    }
};
//---------------------------------------------------------------------------------------------------------------------------------------

class EventLoop
{
private:
    using Functor = std::function<void()>;
    std::thread::id _thread_id; // 线程ID
    int _event_fd;              // eventfd唤醒IO事件监控可能导致的阻塞
    Poller _poller;             // 进行所有描述符的事件监控
    std::unique_ptr<Channel> _event_channel;
    std::vector<Functor> _tasks; // 任务池
    std::mutex _mutex;           // 实现任务池操作的线程安全
    TimeWheel _timer_wheel;

public:
    // 执行任务池中的所有任务
    void RunAllTask()
    {
        std::vector<Functor> functor;
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _tasks.swap(functor);
        }
        for (auto &f : functor)
        {
            f();
        }
        return;
    }
    static int CreateEventFd()
    {
        int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        if (efd < 0)
        {
            ERR_LOG("CREATE EVENTFD FAILED!!");
            abort();
        }
        return efd;
    }
    void ReadEventfd()
    {
        uint64_t res = 0;
        int ret = read(_event_fd, &res, sizeof(res));
        if (ret < 0)
        { // EINTER 表示被信号打断  EAGAIN 表示无数据可读
            if (errno == EINTR)
            {
                return;
            }
            ERR_LOG("READ EVENTFD FAILED");
            abort();
        }
        return;
    }
    void WakeUpEventFd()
    {
        uint64_t val = 1;
        int ret = write(_event_fd, &val, sizeof(val));
        if (ret < 0)
        {
            if (errno == EINTR)
            {
                return;
            }
            ERR_LOG("READ EVENTFD FAILED!");
            abort();
        }
        return;
    }

public:
    EventLoop()
        : _thread_id(std::this_thread::get_id()),
          _event_fd(CreateEventFd()), _poller(), _event_channel(new Channel(this, _event_fd)), _timer_wheel(this)
    {
        _event_channel->SetReadCallBack(std::bind(&EventLoop::ReadEventfd, this));

        // 启动eventfd的读事件监控
        _event_channel->EnableRead();
    }

    void RunInLoop(const Functor &cb)
    { // 当前任务在当前线程即执行，不是则压入队列
        if (IsInLoop())
        {
            return cb();
        }
        return QueueInLoop(cb);
    }

    void QueueInLoop(const Functor &cb)
    { // 将操作压入任务池
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _tasks.push_back(cb);
        }
        // 唤醒有可能以为没事情就绪，导致的epoll阻塞
        // 其实就是给eventfd写一个数据，eventfd就会触发可读事件
        WakeUpEventFd();
    }
    bool IsInLoop()
    { // 判断当前线程是否是EventLoop对应的线程
        return _thread_id == std::this_thread::get_id();
    }
    void AssertInLoop(){
        assert(_thread_id == std::this_thread::get_id());
    }
    void UpdateEvent(Channel *channel)
    { // 添加/修改描述符的事件监控
        _poller.UpDateEvent(channel);
    }
    void RemoveEvent(Channel *channel)
    { // 移除监控
        _poller.ReMoveEvent(channel);
    }
    void TimerAdd(uint64_t id, uint32_t delay, const TaskFunc &cb)
    {
        return _timer_wheel.TimerAdd(id, delay, cb);
    }
    void TimerRefresh(uint64_t id)
    {
        return _timer_wheel.TimerRefresh(id);
    }
    void TimerCancle(uint64_t id)
    {
        return _timer_wheel.TimerCancle(id);
    }
    bool HasTimer(uint64_t id)
    {
        return _timer_wheel.HashTimer(id);
    }

    void Start()
    { // 事件监控 --- 就绪事件处理  ---- 执行任务
        // 1.事件监控
        while(1){
           std::vector<Channel *> actives;
        _poller.Poll(&actives);

        // 2.事件处理
        for (auto &channel : actives)
        {
            channel->HandleEvent();
        }
        // 3.执行任务
        RunAllTask(); 
        }
        
    }
};

//------------------------------------------------------------------------------------------------------------------
class LoopThread{
    private:       //用于实现_loop获取的同步关系，避免线程创建了，但是_loop还没有实例化之前去获取_loop
        std::mutex _mutex; //互斥锁
        std::condition_variable _cond; //条件变量
        EventLoop * _loop;  //EventLoop指针变量，这个对象需要在线程内实例化
        std::thread _thread; // EventLoop对应的线程
    private:
    //实例化EventLoop对象，并且开始运行EventLoop模块的功能  唤醒_cond上可能阻塞的线程
    void ThreadEntry(){
        EventLoop loop;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _loop = &loop;
            _cond.notify_all();
        }
        loop.Start();
    }

    public:
        //创建线程，设定线程入口函数
        LoopThread():_loop(NULL),_thread(std::thread(&LoopThread::ThreadEntry,this)){
        };
        EventLoop* GetLoop(){
            //返回当前线程关联的EventLoop对象指针
            EventLoop* loop = NULL;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _cond.wait(lock,[&](){return _loop!=NULL;}); //_loop为空就一直阻塞
                loop = _loop;
            }
            return loop;
        }
};
//--------------------------------------------------------------------------------------------------------------
class LoopThreadPool{
    private:
        int _thread_count;
        int _next_idx;
        EventLoop* _baseloop;
        std::vector<LoopThread*> _threads;
        std::vector<EventLoop*> _loops;
    public:
        LoopThreadPool(EventLoop* baseloop):_thread_count(0),_next_idx(0),_baseloop(baseloop){
        }
        void SetThreadCount(int count){
            _thread_count = count;
        }
        void Create(){
            if(_thread_count>0){
                _threads.resize(_thread_count);
                _loops.resize(_thread_count);
                for(int i=0;i<_thread_count;++i){
                    _threads[i] = new LoopThread();
                    _loops[i] = _threads[i]->GetLoop();
                }
            }
            return;
        }
        EventLoop* NextLoop(){
            if(_thread_count == 0){
                return _baseloop;
            }
            _next_idx = (_next_idx+1)%_thread_count;   //循环
            return  _loops[_next_idx];
        }
};



//---------------------------------------------------------------------------------------------------------------------------------
enum ConnStatu{
    DISCONNECTED,   //连接关闭状态
    CONNECTING,        // 连接建立成功 - 待处理状态
    CONNECTED,      //连接建立完成，各设置已完成 可以通信
    DISCONNCETING //待关闭状态
};
class Connection;
using PtrConnection = std::shared_ptr<Connection>;

class Connection: public std::enable_shared_from_this<Connection>{
private:
    uint64_t _con_id; //连接的唯一ID，便于连接管理和查找
    //uint64_t _timer_id; //定时器ID，必须唯一 为了简化操作，直接使用_con_id作为定时器id.
    int _sockfd; //连接关联的文件描述符
    bool _enable_inactive_release; //连接是否启动非活跃销毁的判断标志
    EventLoop* _loop; //连接所关联的一个Eventloop
    Socket _socket;  //套接字操作管理
    Channel _channel;  //连接事件管理
    Buffer _in_buffer;  //输入缓冲区    --- 存放从socket中读取到的数据
    Buffer _out_buffer;  //输出缓冲区  --- 存放发送给对端的数据
    Any _contex;    //请求的接收上下文
    ConnStatu _statu;  //连接的状态

    //这四个回调函数是由组件使用者来设置的
    using ConnectedCallback = std::function<void(const PtrConnection&)>;
    using MessageCallback = std::function<void(const PtrConnection&,Buffer*)>;
    using ClosedCallback = std::function<void(const PtrConnection&)>;
    using AnyEventCallback = std::function<void(const PtrConnection&)>;
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _close_callback;
    AnyEventCallback _event_callback;
    //组件内的连接关闭回调
    ClosedCallback _server_closed_callback;
private:
    //五个channel的事件回调函数
    void HandlerRead(){   //描述符可读事件触发后调用的函数
        char buf[65536];
        ssize_t ret = _socket.NonBlockRecv(buf,65535);
        if(ret<0){  
            return ShutDownInLoop();
            //return;
        }else if(ret == 0){
            return;     //这里的等于0是未读取到数据   而非连接断开 
        }
        //将数据放入输入缓冲区
        _in_buffer.WriteAndPush(buf,ret);
        //2.调用message_callback 进行业务处理
        if(_in_buffer.ReadAbleSize()>0){
            return _message_callback(shared_from_this(),&_in_buffer);
        }
        return ReleaseInLoop();  //实际的关闭操作

    }
    void HandlerWrite(){  //描述符触发写事件
        //out_buff中保存的数据就是要发送的数据
        ssize_t ret = _socket.NonBlockSend(_out_buffer.ReadPosition(),_out_buffer.ReadAbleSize());
        if(ret<0){
            //发送错误该关闭连接了
            if(_in_buffer.ReadAbleSize() > 0){
                _message_callback(shared_from_this(),&_in_buffer);
                return ReleaseInLoop(); //这时候就是实际的关闭释放操作了
            }
        }
        _out_buffer.MoveReadOffset(ret);  //记得将读偏移向后移动
        if(_out_buffer.ReadAbleSize() == 0){
            _channel.DisableWrite();  //没有数据待发送就关闭写事件监控
        if(_statu == DISCONNCETING){
        return ReleaseInLoop();
        }
        }
        return;
    }
    void HandlerClose(){   
        //一旦连接挂断，套接字就什么都干不了，因此有数据待处理就处理 完成关闭连接
        if(_in_buffer.ReadAbleSize()>0){
            _message_callback(shared_from_this(),&_in_buffer);
        }
        return ReleaseInLoop();
    }
    void HandlerError(){
        HandlerClose();
    }
    void HandlerEvent(){  // 1.刷新连接活跃度 延迟定时销毁任务   2.调用组件使用者的任意事件回调
        if(_enable_inactive_release == true){
            _loop->TimerRefresh(_con_id);
        }
        if(_event_callback){ 
            _event_callback(shared_from_this());
        }
    }
    void EstablishedInLoop(){   //连接获取之后，给channel设置事件回调 启动读监控
        //1.修改连接状态  2.启动读事件监控 3.调用回调函数
        assert(_statu == CONNECTING);  //状态必定是半连接状态
        _statu = CONNECTED;
        _channel.EnableRead();
        if(_connected_callback){
            _connected_callback(shared_from_this());
        }

    }
    void SendInLoop(const char* data, size_t len){  //并不是实际的发送接口 只是把数据放到发送缓冲区 启动可写事件监控
        if(_statu == DISCONNECTED) return;
        _out_buffer.WriteAndPush(data,len);
        if(_channel.IsWrite() == false){
            _channel.EnableWrite();
        }
    }
    void ReleaseInLoop(){ //这个接口才是实际的释放接口
       // 1.修改连接状态 置为DISCONNECTED  2.移除连接的事件监控 3. 关闭描述符 4.如果当前定时器队列中还有定时销毁任务，则取消任务  5.调用关闭回调函数
        _statu = DISCONNECTED;
        _channel.ReMove();
        _socket.Close();
        
        if(_loop->HasTimer(_con_id)){         //.......................把这个删掉了 额外添加了个特化的
            CancleInactiveRelease();
        }

        if(_close_callback){
            _close_callback(shared_from_this());
        }
        if(_server_closed_callback){  //移除服务器内部管理的连接信息
            _server_closed_callback(shared_from_this());
        }
    }

    void ShutDownInLoop(){      //这个关闭操作并非实际的连接释放操作 需判断还有没有数据待处理    
        _statu == DISCONNCETING;
        if(_in_buffer.ReadAbleSize()>0){
            if(_message_callback){
                _message_callback(shared_from_this(),&_in_buffer);
            }
        }
        if(_out_buffer.ReadAbleSize()>0){   //要么写入数据时出错关闭，要么没数据发送直接关闭
            if(_channel.IsWrite() == false){
                _channel.EnableWrite();
            }
        }
        if(_out_buffer.ReadAbleSize() == 0){
            ReleaseInLoop();
        }
    }
    void EnableInactiveReleaseInLoop(int sec){    //启动超时释放规则
        //1.将判断标准 _enable_inactive_release 置为true
        _enable_inactive_release = true;
        //2.添加定时销毁任务  如果已经存在 刷新延迟即可
        if(_loop->HasTimer(_con_id)){
            return _loop->TimerRefresh(_con_id);
        }
        //3.如果不存在定时销毁任务，则新增
        _loop->TimerAdd(_con_id,sec,std::bind(&Connection::ReleaseTimeInLoop,this));
    }
    void CancleInactiveReleaseInLoop(){
        _enable_inactive_release = false;
        if(_loop->HasTimer(_con_id))
        _loop->TimerCancle(_con_id);
    }
    void UpgradeInLoop(const Any& context,const ConnectedCallback& conn, const MessageCallback& msg, 
                                         const ClosedCallback& closed,const AnyEventCallback& event){   //切换协议 重置上下文数据和阶段处理函数
        SetContext(context);
        _connected_callback = conn;
        _message_callback = msg;
        _close_callback = closed;
        _event_callback = event;
    }
public:
    Connection(EventLoop* loop,uint64_t con_id,int sockfd)
    :_con_id(con_id), _sockfd(sockfd) , _enable_inactive_release(false), _loop(loop),_statu(CONNECTING),_socket(sockfd),_channel(loop,_sockfd){
    _channel.SetCloseCallBack(std::bind(&Connection::HandlerClose,this));
    _channel.SetEventCallBack(std::bind(&Connection::HandlerEvent,this));
    _channel.SetReadCallBack(std::bind(&Connection::HandlerRead,this));
    _channel.SetWriteCallBack(std::bind(&Connection::HandlerWrite,this));
    _channel.SetErrorCallBack(std::bind(&Connection::HandlerError,this));
    }
    ~Connection(){
        DBG_LOG("RELEASE CONNECTION: %p",this);
    }
    int FD(){
        return _sockfd;
    }
    int ID(){
        return _con_id;
    }
    bool Connected(){
        return (_statu == CONNECTED);
    }
    ConnStatu Statu(){
        return _statu ;
    } //返回状态
    void SetContext(const Any&context){  //设置上下文
        _contex = context;
    }
    Any* GetContext(){   //获取上下文
        return &_contex;
    }
    void SetConnectedCallback(const ConnectedCallback& cb){
        _connected_callback = cb;
    }
    void SetMessageCallback(const MessageCallback& cb){
        _message_callback = cb;
    }
    void SetClosedCallback(const ClosedCallback& cb){
        _close_callback = cb;
    }
    void SetAnyEventCallback(const ConnectedCallback& cb){
        _event_callback = cb;
    }
    void SetServerClosedCallback(const ClosedCallback& cb){
        _server_closed_callback = cb;
    }
    void Established(){    //连接就绪后进行channel回调设置 启动读监控 调用_connected_callback
        _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop,this));
    }
    void Send(const char* data, size_t len){   //发送数据到发送缓冲区，启动写事件监控  
        _loop->RunInLoop(std::bind(&Connection::SendInLoop,this,data,len));
    }
    void Shutdown(){   //关闭接口 --- 并不实际关闭， 需判断是否有数待处理
        _loop->RunInLoop(std::bind(&Connection::ShutDownInLoop,this));
    }
    void EnableInactiveRelease(int sec){  //启动非活跃销毁，并定义多长时间   添加 定时任务
        _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop,this,sec));
    }
    void CancleInactiveRelease(){
        _loop->RunInLoop(std::bind(&Connection::CancleInactiveReleaseInLoop,this));
    }
    void ReleaseTimeInLoop(){
        _loop->RunInLoop(std::bind(&Connection::ReleaseInLoop,this));
    }
    void Upgrade(const Any& context,const ConnectedCallback& conn, const MessageCallback& msg, 
                                         const ClosedCallback& closed,const AnyEventCallback& event){   //切换协议 重置上下文数据和阶段处理函数 -- 非线程安全
    //防备新事件触发后 ，处理的时候切换协议还没执行，导致数据用原协议处理了
        _loop->AssertInLoop();
        _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop,this,context,conn,msg,closed,event));
    }
};

//-------------------------------------------------------------------------------------------------------------------------------------------------------
class Acceptor{
    private:
        Socket _socket; // 用于创建监听套接字
        EventLoop* _loop; //用于对监听套接字进行事件监控
        Channel _channel; //用于对监听套接字进行事件管理

        using AcceptCallback = std::function<void(int)>;
        AcceptCallback _accept_callback;
    private:
    //监听套接字的读事件回调处理函数 --- 获取新连接，调用_accept_callback函数进行新连接处理
        void HandRead(){
            int newfd = _socket.Accept();
            if(newfd<0){
                return;
            }
            if(_accept_callback) _accept_callback(newfd);
        }
        int CreateServer(int port){
            bool ret = _socket.CreateServer(port);
            assert(ret == true);
            return _socket.FD();
        }
    public:
    //不能将启动读事件监控放到构造函数中，必须在设置回调函数后，再去启动
    //f
        Acceptor(EventLoop* loop,int port):_socket(CreateServer(port)),_loop(loop),
        _channel(_loop,_socket.FD()){
            _channel.SetReadCallBack(std::bind(&Acceptor::HandRead,this));
        }
        void SetAcceptCallback(const AcceptCallback& cb){
            _accept_callback = cb;
        }
        void Listen(){
            _channel.EnableRead();
        }
};


//---------------------------------------------------------------------------------------------------------------------------

class TcpServer{
    private:
        int _port;
        uint64_t _conn_id; //自动增长的连接Id
        Acceptor _acceptor; //监听套接字的管理对象
        EventLoop  _baseloop; //这是主线程的EventLoop对象，负责监听事件的处理
        LoopThreadPool _pool; //这是从属EventLoop线程池
        int _timeout; //这是非活跃连接的统计事件 -- 多长时间无通信就是非活跃连接
        bool _enable_inactive_release;    //释放需要释放非活跃连接的标志
        std::unordered_map<uint64_t,PtrConnection> _conns; //保存管理所有连接对应的shared_ptr对象；

        using Functor = std::function<void()>;  //用户设置的定时任务
        //这四个回调函数是由组件使用者来设置的
        using ConnectedCallback = std::function<void(const PtrConnection&)>;
        using MessageCallback = std::function<void(const PtrConnection&,Buffer*)>;
        using ClosedCallback = std::function<void(const PtrConnection&)>;
        using AnyEventCallback = std::function<void(const PtrConnection&)>;
        ConnectedCallback _connected_callback;
        MessageCallback _message_callback;
        ClosedCallback _close_callback;
        AnyEventCallback _event_callback;
    
    private: 
        void NewConnection(int fd){  //给新连接构造一个Connection管理
            ++_conn_id;
            PtrConnection conn(new Connection(_pool.NextLoop(),_conn_id,fd));
            conn->SetMessageCallback(_message_callback);
            conn->SetClosedCallback(_close_callback);
            conn->SetConnectedCallback(_connected_callback);
            conn->SetAnyEventCallback(_event_callback);
            conn->SetServerClosedCallback(std::bind(&TcpServer::RemoveConnection,this,std::placeholders::_1));
            if(_enable_inactive_release) conn->EnableInactiveRelease(_timeout); //启动超时销毁
            conn->Established(); //就绪初始化
            _conns.insert(std::make_pair(_conn_id,conn));
        }
        void RemoveConnectionInLoop(const PtrConnection& conn){
            int id = conn->ID();
            auto it =_conns.find(id);
            if(it != _conns.end()){
                _conns.erase(it);
            }
        }
       
        void RunAfterInLoop(const Functor& task,int delay){
            _conn_id++;
            _baseloop.TimerAdd(_conn_id,delay,task);
        }

    public:
        TcpServer(int port)
        :_port(port),
        _conn_id(0),
        _enable_inactive_release(false)
        ,_acceptor(&_baseloop,_port)
        ,_pool(&_baseloop)
        {
            _acceptor.SetAcceptCallback(std::bind(&TcpServer::NewConnection,this,std::placeholders::_1));
            _acceptor.Listen(); //将监听套接字挂到baseloop上开始监控
        }
        void SetThreadCount(int count){
            return _pool.SetThreadCount(count);
        }
        void SetConnectedCallback(const ConnectedCallback& cb){
        _connected_callback = cb;
        }
        void SetMessageCallback(const MessageCallback& cb){
            _message_callback = cb;
        }
        void SetClosedCallback(const ClosedCallback& cb){
            _close_callback = cb;
        }
        void SetAnyEventCallback(const ConnectedCallback& cb){
            _event_callback = cb;
        }
        void EnableInactiveRelease(int timeout){
            _timeout = timeout;
            _enable_inactive_release = true;
        }
        void RunAfter(const Functor& task,int delay){
            _baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop,this,task,delay));
        }
         void RemoveConnection(const PtrConnection& conn){  //从管理Connection的_conns中移除连接信息
            _baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop,this,conn));
        }
        void Start(){
            _pool.Create(); //创建线程池中的从属线程
            return _baseloop.Start();
        }

};








void Channel::UpDate() { return _loop->UpdateEvent(this); }
// 移除监控
void Channel::ReMove() { return _loop->RemoveEvent(this); }

void TimeWheel::TimerAdd(uint64_t id, uint32_t delay, const TaskFunc &cb)
{
    _loop->RunInLoop(std::bind(&TimeWheel::TimerAddInLoop, this, id, delay, cb));
}
void TimeWheel::TimerRefresh(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimeWheel::TimerRefreshInLoop, this, id));
}
void TimeWheel::TimerCancle(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimeWheel::TimerCancleInLoop, this, id));
}

class NetWork{
    public:
        NetWork(){
            signal(SIGPIPE,SIG_IGN);
        }
};
static NetWork nw;